package com.desheng.bigdata.flink.stream.source

import java.util.concurrent.atomic.AtomicInteger

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

/**
  * 用户自定义source function之并行的sourcefunction
  */
object _05UserDefineRichParallelSourceFromIncrementNumber {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment

        val num = env.addSource(new MyParallelSourceFunction()).setParallelism(2)

        num.print()

        env.execute(s"${_05UserDefineRichParallelSourceFromIncrementNumber.getClass.getSimpleName}")
    }
}

/**
  * 模拟不断增长的数字
  * 当并行度被设置为2，我们看到，同一个数字被消费了两次，原因在于由于并行度是2，也就是意味着有两个task
  * 所以该MyParallelSourceFunction类被这两个task各实例化了一次，进而我们看到了同一个数字被消费了两次的现象
  */
class MyRichParallelSourceFunction extends RichParallelSourceFunction[Int] {

    var counter: AtomicInteger = new AtomicInteger(0)
    override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {
//        counter = new AtomicInteger(0)
        while(true) {
            ctx.collect(counter.getAndIncrement())
            Thread.sleep(1000)
        }
    }

    override def cancel(): Unit = {
        if(counter != null) {
            counter = new AtomicInteger(0)
        }
    }
}
